package drds.binlog.parse.mysql.table_meta_data;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastsql.sql.repository.Schema;
import drds.binlog.common.Authors;
import drds.binlog.common.position.EntryPosition;
import drds.binlog.database_driver.packets.server.ResultSetPacket;
import drds.binlog.event_filter.EventFilter;
import drds.binlog.parse.FieldMetaData;
import drds.binlog.parse.TableMetaData;
import drds.binlog.parse.exception.ParseException;
import drds.binlog.parse.mysql.Connection;
import drds.binlog.parse.mysql.dbsync.TableMetaDataCache;
import drds.binlog.parse.mysql.ddl.Ddl;
import drds.binlog.parse.mysql.ddl.FastsqlDdlParser;
import drds.binlog.parse.mysql.table_meta_data.dao.MetaHistoryDAO;
import drds.binlog.parse.mysql.table_meta_data.dao.MetaHistoryDo;
import drds.binlog.parse.mysql.table_meta_data.dao.MetaSnapshotDAO;
import drds.binlog.parse.mysql.table_meta_data.dao.MetaSnapshotDO;
import drds.common.Author;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.MDC;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;


@Slf4j
@Author(name = Authors.LI_YANG)
public class BaseOnDatabaseTableMetaDataGetter implements TableMetaDataGetter
{

    public static final EntryPosition INIT_POSITION = new EntryPosition(-1L, "0", 0L, -2L);

    private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'");
    private static Pattern h2Pattern = Pattern.compile("Unique index or primary key violation");


    @Setter
    @Getter
    private String destination;
    @Setter
    @Getter
    private BaseOnMemoryTableMetaDataGetter baseOnMemoryTableMetaDataGetter;
    @Setter
    @Getter
    private Connection connection;

    // 查询meta信息的链接
    @Setter
    @Getter
    private EventFilter eventFilter;
    @Setter
    @Getter
    private EventFilter blackFilter;
    @Setter
    @Getter
    private EntryPosition lastEntryPosition;
    @Setter
    @Getter
    private ScheduledExecutorService scheduledExecutorService;
    @Setter
    @Getter
    private MetaHistoryDAO metaHistoryDAO;
    @Setter
    @Getter
    private MetaSnapshotDAO metaSnapshotDAO;

    public BaseOnDatabaseTableMetaDataGetter()
    {

    }

    @Override
    public boolean init(final String destination)
    {
        this.destination = destination;
        this.baseOnMemoryTableMetaDataGetter = new BaseOnMemoryTableMetaDataGetter();
        this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory()
        {

            @Override
            public Thread newThread(Runnable runnable)
            {
                Thread thread = new Thread(runnable, "[scheduledExecutorService-table-metadata-snapshot]");
                thread.setDaemon(true);
                return thread;
            }
        });

        // 24小时生成一份snapshot
        scheduledExecutorService.scheduleWithFixedDelay(new Runnable()
        {

            @Override
            public void run()
            {
                try
                {
                    MDC.put("destination", destination);
                    applySnapshotToDB(lastEntryPosition, false);
                } catch (Throwable e)
                {
                    log.error("scheudle applySnapshotToDB faield", e);
                }
            }
        }, 24, 24, TimeUnit.HOURS);
        return true;
    }

    @Override
    public TableMetaData find(String schemaName, String tableName)
    {
        synchronized (baseOnMemoryTableMetaDataGetter)
        {
            return baseOnMemoryTableMetaDataGetter.find(schemaName, tableName);
        }
    }

    @Override
    public boolean apply(EntryPosition entryPosition, String schemaName, String ddl, String extra)
    {
        // 首先记录到内存结构
        synchronized (baseOnMemoryTableMetaDataGetter)
        {
            if (baseOnMemoryTableMetaDataGetter.apply(entryPosition, schemaName, ddl, extra))
            {
                this.lastEntryPosition = entryPosition;
                // 同步每次变更给远程做历史记录
                return saveToDataBase(entryPosition, schemaName, ddl, extra);
            } else
            {
                throw new RuntimeException("apply to memory is failed");
            }
        }
    }

    @Override
    public boolean rollback(EntryPosition position)
    {
        // 每次rollback需要重新构建一次memory data
        this.baseOnMemoryTableMetaDataGetter = new BaseOnMemoryTableMetaDataGetter();
        boolean flag = false;
        EntryPosition entryPosition = buildEntryPositionFromSnapshot(position);
        if (entryPosition != null)
        {
            applyHistoryOnMemory(entryPosition, position);
            flag = true;
        }

        if (!flag)
        {
            // 如果没有任何数据，则为初始化状态，全量dump一份关注的表
            if (dumpTableMeta(connection, eventFilter))
            {
                // 记录一下snapshot结果,方便快速恢复
                flag = applySnapshotToDB(INIT_POSITION, true);
            }
        }

        return flag;
    }

    @Override
    public Map<String, String> snapshot()
    {
        return baseOnMemoryTableMetaDataGetter.snapshot();
    }

    /**
     * 初始化的时候dump一下表结构
     */
    private boolean dumpTableMeta(Connection connection, final EventFilter filter)
    {
        try
        {
            ResultSetPacket packet = connection.query("show databases");
            List<String> schemas = new ArrayList<String>();
            for (String schema : packet.getFieldValueList())
            {
                schemas.add(schema);
            }

            for (String schema : schemas)
            {
                packet = connection.query("show tables from `" + schema + "`");
                List<String> tables = new ArrayList<String>();
                for (String table : packet.getFieldValueList())
                {
                    String fullName = schema + "." + table;
                    if (blackFilter == null || !blackFilter.match(fullName))
                    {
                        if (filter == null || filter.match(fullName))
                        {
                            tables.add(table);
                        }
                    }
                }

                if (tables.isEmpty())
                {
                    continue;
                }

                StringBuilder sql = new StringBuilder();
                for (String table : tables)
                {
                    sql.append("show create table `" + schema + "`.`" + table + "`;");
                }

                List<ResultSetPacket> packets = connection.querys(sql.toString());
                for (ResultSetPacket onePacket : packets)
                {
                    if (onePacket.getFieldValueList().size() > 1)
                    {
                        String oneTableCreateSql = onePacket.getFieldValueList().get(1);
                        baseOnMemoryTableMetaDataGetter.apply(INIT_POSITION, schema, oneTableCreateSql, null);
                    }
                }
            }

            return true;
        } catch (IOException e)
        {
            throw new ParseException(e);
        }
    }

    private boolean saveToDataBase(EntryPosition entryPosition, String schemaName, String ddl, String extra)
    {
        Map<String, String> map = new HashMap<String, String>();
        map.put("destination", destination);
        map.put("binlogFile", entryPosition.getJournalName());
        map.put("binlogOffest", String.valueOf(entryPosition.getPosition()));
        map.put("binlogMasterId", String.valueOf(entryPosition.getServerId()));
        map.put("binlogTimestamp", String.valueOf(entryPosition.getTimestamp()));
        map.put("useSchema", schemaName);
        if (map.isEmpty())
        {
            throw new RuntimeException("apply failed caused by content is empty in saveToDataBase");
        }
        // 待补充
        List<Ddl> ddlList = FastsqlDdlParser.parse(ddl, schemaName);
        if (ddlList.size() > 0)
        {
            Ddl ddl1 = ddlList.get(0);
            map.put("sqlSchema", ddl1.getSchemaName());
            map.put("sqlTable", ddl1.getTableName());
            map.put("sqlType", ddl1.getEventType().name());
            map.put("sqlText", ddl);
            map.put("extra", extra);
        }

        MetaHistoryDo metaHistoryDo = new MetaHistoryDo();
        try
        {
            BeanUtils.populate(metaHistoryDo, map);
            // 会建立唯一约束,解决:
            // 1. 重复的binlog file+offest
            // 2. 重复的masterId+timestamp
            metaHistoryDAO.insert(metaHistoryDo);
        } catch (Throwable e)
        {
            if (isUkDuplicateException(e))
            {
                // 忽略掉重复的位点
                log.warn("dup apply for sql : " + ddl);
            } else
            {
                throw new ParseException("apply history to db failed caused by : " + e.getMessage(), e);
            }

        }
        return true;
    }

    /**
     * 发布数据到console上
     */
    private boolean applySnapshotToDB(EntryPosition position, boolean init)
    {
        // 获取一份快照
        BaseOnMemoryTableMetaDataGetter baseOnMemoryTableMetaDataGetter = new BaseOnMemoryTableMetaDataGetter();
        Map<String, String> schemaDdls = null;
        synchronized (this.baseOnMemoryTableMetaDataGetter)
        {
            if (!init && position == null)
            {
                // 如果是持续构建,则识别一下是否有DDL变更过,如果没有就忽略了
                return false;
            }
            schemaDdls = this.baseOnMemoryTableMetaDataGetter.snapshot();
            for (Map.Entry<String, String> entry : schemaDdls.entrySet())
            {
                baseOnMemoryTableMetaDataGetter.apply(position, entry.getKey(), entry.getValue(), null);
            }
        }

        // 基于临时内存对象进行对比
        boolean compareAll = true;
        for (Schema schema : baseOnMemoryTableMetaDataGetter.getSchemaRepository().getSchemas())
        {
            for (String table : schema.showTables())
            {
                if (!compareTableMetaDbAndMemory(connection, baseOnMemoryTableMetaDataGetter, schema.getName(), table))
                {
                    compareAll = false;
                }
            }
        }
        if (compareAll)
        {
            Map<String, String> map = new HashMap<String, String>();
            map.put("destination", destination);
            map.put("binlogFile", position.getJournalName());
            map.put("binlogOffest", String.valueOf(position.getPosition()));
            map.put("binlogMasterId", String.valueOf(position.getServerId()));
            map.put("binlogTimestamp", String.valueOf(position.getTimestamp()));
            map.put("data", JSON.toJSONString(schemaDdls));
            if (map.isEmpty())
            {
                throw new RuntimeException("apply failed caused by content is empty in applySnapshotToDB");
            }

            MetaSnapshotDO metaSnapshotDO = new MetaSnapshotDO();
            try
            {
                BeanUtils.populate(metaSnapshotDO, map);
                metaSnapshotDAO.insert(metaSnapshotDO);
            } catch (Throwable e)
            {
                if (isUkDuplicateException(e))
                {
                    // 忽略掉重复的位点
                    log.info("dup apply snapshot use setIndexAsOriginIndexAndOffset$originIndex : " + position + " , just ignore");
                } else
                {
                    throw new ParseException("apply failed caused by : " + e.getMessage(), e);
                }
            }
            return true;
        } else
        {
            log.error("compare failed , check log");
        }
        return false;
    }

    private boolean compareTableMetaDbAndMemory(Connection connection, BaseOnMemoryTableMetaDataGetter baseOnMemoryTableMetaDataGetter,
                                                final String schemaName, final String tableName)
    {
        TableMetaData tableMetaDataFromMemory = baseOnMemoryTableMetaDataGetter.find(schemaName, tableName);

        TableMetaData tableMetaData = new TableMetaData();
        tableMetaData.setDatabaseName(schemaName);
        tableMetaData.setTableName(tableName);
        String createTableDdl = null;
        try
        {
            ResultSetPacket resultSetPacket = connection.query("show create table " + getFullName(schemaName, tableName));
            if (resultSetPacket.getFieldValueList().size() > 1)
            {
                createTableDdl = resultSetPacket.getFieldValueList().get(1);
                tableMetaData.setFieldMetaDataList(TableMetaDataCache.parseTableMeta(schemaName, tableName, resultSetPacket));
            }
        } catch (Throwable e)
        {
            try
            {

                connection.reconnect();
                ResultSetPacket resultSetPacket = connection.query("show create table " + getFullName(schemaName, tableName));
                if (resultSetPacket.getFieldValueList().size() > 1)
                {
                    createTableDdl = resultSetPacket.getFieldValueList().get(1);
                    tableMetaData.setFieldMetaDataList(TableMetaDataCache.parseTableMeta(schemaName, tableName, resultSetPacket));
                }
            } catch (IOException e1)
            {
                if (e.getMessage().contains("errorNumber=1146"))
                {
                    log.error("table not exist in db , pls check :" + getFullName(schemaName, tableName) + " , mem : "
                            + tableMetaDataFromMemory);
                    return false;
                }
                throw new ParseException(e);
            }
        }

        boolean result = compareTableMetaData(tableMetaDataFromMemory, tableMetaData);
        if (!result)
        {
            log.error("pls submit github issue, show create table ddl:" + createTableDdl + " , compare failed . \n db : "
                    + tableMetaData + " \n mem : " + tableMetaDataFromMemory);
        }
        return result;
    }

    private EntryPosition buildEntryPositionFromSnapshot(EntryPosition position)
    {
        try
        {
            MetaSnapshotDO metaSnapshotDO = metaSnapshotDAO.findByTimestamp(destination, position.getTimestamp());
            if (metaSnapshotDO == null)
            {
                return null;
            }
            String binlogFileName = metaSnapshotDO.getBinlogFileName();
            Long binlogFileOffest = metaSnapshotDO.getBinlogFileOffest();
            String binlogMasterId = metaSnapshotDO.getBinlogMasterId();
            Long binlogTimestamp = metaSnapshotDO.getBinlogTimestamp();

            EntryPosition entryPosition = new EntryPosition(Long.valueOf(binlogMasterId == null ? "-2" : binlogMasterId), binlogFileName,
                    binlogFileOffest == null ? 0l : binlogFileOffest,
                    binlogTimestamp == null ? 0l : binlogTimestamp
            );
            // data存储为Map<String,String>，每个分库一套建表
            String sqlData = metaSnapshotDO.getData();
            JSONObject jsonObj = JSON.parseObject(sqlData);
            for (Map.Entry entry : jsonObj.entrySet())
            {
                // 记录到内存
                if (!baseOnMemoryTableMetaDataGetter.apply(entryPosition,
                        ObjectUtils.toString(entry.getKey()),
                        ObjectUtils.toString(entry.getValue()),
                        null))
                {
                    return null;
                }
            }

            return entryPosition;
        } catch (Throwable e)
        {
            throw new ParseException("apply failed caused by : " + e.getMessage(), e);
        }
    }

    private boolean applyHistoryOnMemory(EntryPosition position, EntryPosition rollbackPosition)
    {
        try
        {
            List<MetaHistoryDo> metaHistoryDoList = metaHistoryDAO.findByTimestamp(destination,
                    position.getTimestamp(),
                    rollbackPosition.getTimestamp());
            if (metaHistoryDoList == null)
            {
                return true;
            }

            for (MetaHistoryDo metaHistoryDo : metaHistoryDoList)
            {
                String binlogFile = metaHistoryDo.getBinlogFile();
                Long binlogOffest = metaHistoryDo.getBinlogOffest();
                String binlogMasterId = metaHistoryDo.getBinlogMasterId();
                Long binlogTimestamp = metaHistoryDo.getBinlogTimestamp();
                String useSchema = metaHistoryDo.getUseSchema();
                String sqlData = metaHistoryDo.getSqlText();
                EntryPosition entryPosition = new EntryPosition(Long.valueOf(binlogMasterId == null ? "-2" : binlogMasterId), binlogFile,
                        binlogOffest == null ? 0L : binlogOffest,
                        binlogTimestamp == null ? 0L : binlogTimestamp
                );

                // 如果是同一秒内,对比一下history的位点，如果比期望的位点要大，忽略之
                if (entryPosition.getTimestamp() > rollbackPosition.getTimestamp())
                {
                    continue;
                } else if (rollbackPosition.getServerId() == entryPosition.getServerId()
                        && entryPosition.compareTo(rollbackPosition) > 0)
                {
                    continue;
                }

                // 记录到内存
                if (!baseOnMemoryTableMetaDataGetter.apply(entryPosition, useSchema, sqlData, null))
                {
                    return false;
                }

            }

            return metaHistoryDoList.size() > 0;
        } catch (Throwable e)
        {
            throw new ParseException("apply failed", e);
        }
    }

    private String getFullName(String schema, String table)
    {
        StringBuilder builder = new StringBuilder();
        return builder.append('`')
                .append(schema)
                .append('`')
                .append('.')
                .append('`')
                .append(table)
                .append('`')
                .toString();
    }

    private boolean compareTableMetaData(TableMetaData source, TableMetaData target)
    {
        if (!StringUtils.equalsIgnoreCase(source.getDatabaseName(), target.getDatabaseName()))
        {
            return false;
        }

        if (!StringUtils.equalsIgnoreCase(source.getTableName(), target.getTableName()))
        {
            return false;
        }

        List<FieldMetaData> sourceFieldMetaDataList = source.getFieldMetaDataList();
        List<FieldMetaData> targetFieldMetaDataList = target.getFieldMetaDataList();
        if (sourceFieldMetaDataList.size() != targetFieldMetaDataList.size())
        {
            return false;
        }

        for (int i = 0; i < sourceFieldMetaDataList.size(); i++)
        {
            FieldMetaData sourceFieldMetaData = sourceFieldMetaDataList.get(i);
            FieldMetaData targetFieldMetaData = targetFieldMetaDataList.get(i);
            if (!StringUtils.equalsIgnoreCase(sourceFieldMetaData.getColumnName(), targetFieldMetaData.getColumnName()))
            {
                return false;
            }

            if (!StringUtils.equalsIgnoreCase(sourceFieldMetaData.getColumnType(), targetFieldMetaData.getColumnType()))
            {
                return false;
            }

            if (!StringUtils.equalsIgnoreCase(sourceFieldMetaData.getDefaultValue(), targetFieldMetaData.getDefaultValue()))
            {
                return false;
            }

            if (sourceFieldMetaData.isNullable() != targetFieldMetaData.isNullable())
            {
                return false;
            }

            // mysql会有一种处理,针对show create只有uk没有pk时，会在desc默认将uk当做pk
            if (//

                    (sourceFieldMetaData.isPrimaryKey() || sourceFieldMetaData.isUniqueKey())//
                            !=//
                            (targetFieldMetaData.isPrimaryKey() || targetFieldMetaData.isUniqueKey())//
                    )//
            {
                return false;
            }
        }

        return true;
    }


    public boolean isUkDuplicateException(Throwable t)
    {
        if (pattern.matcher(t.getMessage()).find() || h2Pattern.matcher(t.getMessage()).find())
        {
            // 违反外键约束时也抛出这种异常，所以这里还要判断包含字符串Duplicate entry
            return true;
        }
        return false;
    }
}
